package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.utils.MykafkaUtil;
import com.ibm.icu.text.SimpleDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Date;

public class OrderWideApp {
    public static void main(String[] args) throws Exception {
        // TODO: 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度并读取kafka分区数据
        env.setParallelism(4);
        // TODO: 2. 设置检查点 略
        // TODO  3. 从Kafka中读取数据
        // 3.1 声明主题以及消费者组
        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwm_order_detail";
        String ordrWineSinkTopic = "dwm_order_wide";
        String groupId = "order_wide_group";
        // 3.2 获取kafkaSource
        FlinkKafkaConsumer<String> sourdeOrderInfo = MykafkaUtil.getkafkaSource(orderInfoSourceTopic, groupId);
        FlinkKafkaConsumer<String> sourceOrderDetail = MykafkaUtil.getkafkaSource(orderDetailSourceTopic, groupId);
        // 3.3 读取数据封装成流
        DataStreamSource<String> orderInfoJsonDS = env.addSource(sourdeOrderInfo);
        DataStreamSource<String> orderDetailJsonStrIDS = env.addSource(sourceOrderDetail);
        // 3.4 对订单流中的数据进行结构转化 jsonStr -> 实体类对象
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = orderInfoJsonDS.map(
                new RichMapFunction<String, OrderInfo>() {
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderInfo map(String jsonStr) throws Exception {
                        // 将json字符串转为订单实体类
                        OrderInfo orderInfo = JSON.parseObject(jsonStr, OrderInfo.class);
                        //
                        String create_time = orderInfo.getCreate_time();
                        long time = sdf.parse(create_time).getTime();
                        orderInfo.setCreate_ts(time);
                        return orderInfo;
                    }
                });
        // 3.5 对订单明细流中的数据进行结构转换 jsonStr -->实体类对象
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = orderDetailJsonStrIDS.map(
                new RichMapFunction<String, OrderDetail>() {
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }

                    @Override
                    public OrderDetail map(String jsonstr) throws Exception {
                        OrderDetail oderDetail = JSON.parseObject(jsonstr, OrderDetail.class);
                        String create_time = oderDetail.getCreate_time();
                        Date parse = sdf.parse(create_time);
                        long time = parse.getTime();
                        oderDetail.setCreate_ts(time);
                        return oderDetail;
                    }
                }
        );
        orderInfoDS.print(">>>");
        orderDetailDS.print("<<<<");
        // TODO: 4. 指定Watermark以及提取事件时间字段
        // 4.1 订单
        SingleOutputStreamOperator<OrderInfo> orderInfoWithmarkDS = orderInfoDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderInfo>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                            @Override
                            public long extractTimestamp(OrderInfo orderInfo, long recordTimestamp) {
                                return orderInfo.getCreate_ts();
                            }
                        })
        );
        // 4.2 订单明细
        SingleOutputStreamOperator<OrderDetail> orderDetailWithWatermarkDS = orderDetailDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<OrderDetail>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
                            @Override
                            public long extractTimestamp(OrderDetail oderDetail, long recordTimestamp) {
                                return oderDetail.getCreate_ts();
                            }
                        })
        );

        // TODO: 5. 按照订单Id进行分组
        KeyedStream<OrderInfo, Long> orderInfokeyedDS = orderInfoWithmarkDS.keyBy(orderInfo -> orderInfo.getId());
        KeyedStream<OrderDetail, Long> orderDetailkeyedDS = orderDetailDS.keyBy(r -> r.getId());
        // TODO: 6. 订单和订单明细进行双流join(intervalJoin)
        SingleOutputStreamOperator<OrderWide> orderWideDS = orderInfokeyedDS
                .intervalJoin(orderDetailkeyedDS)
                .between(Time.seconds(-5), Time.seconds(5))
                .process(
                        new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                            @Override
                            public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context ctx, Collector<OrderWide> out) throws Exception {
                                out.collect(new OrderWide(orderInfo, orderDetail));
                            }
                        }
                );

        orderWideDS.print();
        env.execute();
    }
}
